package com.yss.netty.handler;

import com.yss.netty.processor.AbstractHandlerProcessor;
import com.yss.netty.protocol.CommandType;
import com.yss.netty.protocol.Message;
import com.yss.netty.protocol.Pong;
import com.yss.netty.protocol.SerType;
import com.yss.netty.util.Host;
import com.yss.netty.util.Pair;
import com.yss.netty.util.serialize.ISerialize;
import com.yss.netty.util.serialize.JsonSerializeUtil;
import com.yss.pojo.NettyServerAck;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.timeout.IdleStateEvent;
import lombok.extern.slf4j.Slf4j;

import java.net.InetSocketAddress;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;


@Slf4j
@ChannelHandler.Sharable
public class NettyServerHandler extends ChannelInboundHandlerAdapter {

    /**
     * Sub-service provides the processor corresponding to the CommandType
     */
    private final static ConcurrentHashMap<CommandType, Pair<AbstractHandlerProcessor, ExecutorService>> processors;

    static {
        processors = new ConcurrentHashMap<>();
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        Message message = (Message) msg;
        Channel channel = ctx.channel();
        InetSocketAddress socketAddress = (InetSocketAddress) channel.remoteAddress();

        log.info("\nReceive Message : {} from : {}", new String(message.getBody()), Host.of(socketAddress).getAddress());

        if (message.getCommandType() == CommandType.PING) {
            channel.writeAndFlush(Pong.pong());
        } else {
            processReceivedMessage(channel, message);
        }
    }

    /**
     * Asynchronous execution of associated sub-components
     * <p>
     * For the Pipeline of the same Channel, it is blocking processing
     *
     * @param channel
     * @param message
     */
    private void processReceivedMessage(final Channel channel, final Message message) {
        try {
            CommandType commandType = message.getCommandType();
            Pair<AbstractHandlerProcessor, ExecutorService> processorPair = processors.get(commandType);

            if (processorPair != null) {
                Runnable r = () -> {
                    Object originalMsg = deserialization(message);
                    processorPair.getLeft().execute(commandType, channel, message.getOpaque(), originalMsg);
                };

                processorPair.getRight().execute(r);
            } else {
                this.sendAckWhenError(channel, String.format("No AbstractHandlerProcessor for commandType : %s", commandType.name()), message.getOpaque());
                log.error("\nNo AbstractHandlerProcessor for commandType : {}", commandType.name());
            }

        } catch (Exception e) {
            log.error("\nServer Handler failed : {}", e.getMessage());
        }
    }

    /**
     * When no processor for a commandType
     * will directly trigger this method
     *
     * @param channel
     * @param message
     * @param opaque
     */
    private void sendAckWhenError(final Channel channel, String message, long opaque) {
        NettyServerAck ack = NettyServerAck.builder().code(-1).msg(message).build();
        ISerialize iSerialize = new JsonSerializeUtil();
        byte[] body = iSerialize.serialize(ack);

        Message ackMessage = Message.of(CommandType.SERVERACK, SerType.JSON, body, opaque);
        channel.writeAndFlush(ackMessage);
    }

    /**
     * According to serType、commandType、body to deserialization data
     *
     * @param message
     * @return
     */
    private Object deserialization(final Message message) {
        ISerialize iSerialize = SerType.matchUtilBySerType(message.getSerType());
        byte[] body = message.getBody();
        CommandType commandType = message.getCommandType();

        return iSerialize.deserialize(body, commandType.getMappingClass());
    }

    /**
     * Register processor
     *
     * @param commandType
     * @param processor
     * @param executorService
     */
    public void registerProcessor(CommandType commandType, AbstractHandlerProcessor processor, ExecutorService executorService) {
        processors.put(commandType, Pair.of(processor, executorService));
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        log.error("\nNettyServerHandler has an exception : {} ", cause.getMessage());
        ctx.channel().close();
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        // https://www.zhihu.com/search?type=content&q=TCP%E8%BF%9E%E6%8E%A5%E6%9A%B4%E6%B6%A8 避免客户端直接断网，导致四次挥手进行不下去
        if (evt instanceof IdleStateEvent) {
            ctx.channel().close();
        } else {
            super.userEventTriggered(ctx, evt);
        }
    }
}
